﻿using System;
using System.Collections.Generic;
using Microsoft.ServiceBus;
using Microsoft.ServiceBus.Messaging;
using System.IO;

namespace AppFabric.ServiceBus.Common
{
    public class ServiceBusHelper: IDisposable
    {
        public ServiceBusNamespace Token { get; set; }
        private NamespaceManager namespaceManager = null;
        private TokenProvider credentials;
        private Uri serviceUri;

        #region Public Methods
        public bool Connect()
        {
            credentials = TokenProvider.CreateSharedSecretTokenProvider(Token.IssuerName, Token.IssuerKey);
            serviceUri = ServiceBusEnvironment.CreateServiceUri("sb", Token.Namespace, Token.ServicePath);
            namespaceManager = new NamespaceManager(serviceUri, credentials);
            return true;
        }

        public IEnumerable<QueueDescription> GetQueues()
        {
            return namespaceManager.GetQueues();
        }

        public IEnumerable<SubscriptionDescription> GetSubscriptions(string topicPath)
        {
            return namespaceManager.GetSubscriptions(topicPath);
        }

        public IEnumerable<TopicDescription> GetTopics()
        {
            return namespaceManager.GetTopics();
        }

        public bool SendMessageToQueue(string message, string queueName, MessageSettings settings)
        {
            MessagingFactory factory = MessagingFactory.Create(serviceUri, credentials);
            QueueClient queueClient = factory.CreateQueueClient(queueName);
            BrokeredMessage payLoad = GetPayload(message, settings);
            queueClient.Send(payLoad);
            return true;
        }

        public bool SendMessageToTopic(string message, string topic, MessageSettings settings)
        {
            MessagingFactory factory = MessagingFactory.Create(serviceUri, credentials);
            TopicClient topicClient = factory.CreateTopicClient(topic);
            BrokeredMessage payLoad = GetPayload(message, settings);
            topicClient.Send(payLoad);
            return true;
        }

        public List<Message> ReceiveQueueMessages(string queue, ReceiveMode mode)
        {
            QueueDescription queueDescription = namespaceManager.GetQueue(queue);
            if (queueDescription != null)
            {
                MessagingFactory factory = MessagingFactory.Create(serviceUri, credentials);
                QueueClient client = factory.CreateQueueClient(queue, mode);
                List<Message> messageList = new List<Message>();

                for (int count = 0; count < queueDescription.MessageCount; count++)
                {
                    BrokeredMessage message = client.Receive(TimeSpan.FromSeconds(5));
                    if (message != null)
                    {
                        if (mode == ReceiveMode.PeekLock)
                            message.Complete();
                        messageList.Add(GetMessage(message));
                    }
                }

                return messageList;
            }
            return null;
        }

        public List<Message> ReceiveSubscriptionMessages(string topic, string subscriptionname, ReceiveMode mode)
        {
            SubscriptionDescription subscriptionDescription = namespaceManager.GetSubscription(topic, subscriptionname);
            if (subscriptionDescription != null)
            {
                MessagingFactory factory = MessagingFactory.Create(serviceUri, credentials);
                SubscriptionClient client = factory.CreateSubscriptionClient(topic, subscriptionname, mode);
                List<Message> messageList = new List<Message>();

                for (int count = 0; count < subscriptionDescription.MessageCount; count++)
                {
                    BrokeredMessage message = client.Receive(TimeSpan.FromSeconds(5));
                    if (message != null)
                    {
                        if (mode == ReceiveMode.PeekLock)
                            message.Complete();
                        messageList.Add(GetMessage(message));
                    }
                }

                return messageList;
            }
            return null;
        }

        public void Dispose()
        {
            namespaceManager = null;
        }
        #endregion

        #region Private Methods
        private BrokeredMessage GetPayload(string message, MessageSettings settings)
        {
            BrokeredMessage payload = new BrokeredMessage(message);
            if (settings != null)
            {
                if (!string.IsNullOrEmpty(settings.MessageId))
                    payload.MessageId = settings.MessageId;
                if (!string.IsNullOrEmpty(settings.ContentType))
                    payload.ContentType = settings.ContentType;
                if (!string.IsNullOrEmpty(settings.CorrelationId))
                    payload.CorrelationId = settings.CorrelationId;
                if (!string.IsNullOrEmpty(settings.Label))
                    payload.Label = settings.Label;
                if (!string.IsNullOrEmpty(settings.SessionId))
                    payload.SessionId = settings.SessionId;
                if (settings.TimeToLive.TotalSeconds > 0)
                    payload.TimeToLive = settings.TimeToLive;
                if (settings.MessageProperties.Count > 0)
                {
                    foreach (var property in settings.MessageProperties)
                    {
                        payload.Properties.Add(property.Key, property.Value);
                    }
                }
            }
            return payload;
        }

        private Message GetMessage(BrokeredMessage message)
        {
            Message msg = new Message();
            msg.ContentType = message.ContentType;
            msg.CorrelationId = message.CorrelationId;
            msg.Label = message.Label;

            Stream messageStream = message.GetBody<Stream>();
            using (StreamReader reader = new StreamReader(messageStream))
            {
                msg.MessageBody = reader.ReadToEnd();
            }
            msg.MessageId = message.MessageId;
            msg.MessageProperties = message.Properties;
            msg.SessionId = message.SessionId;
            msg.TimeToLive = message.TimeToLive;
            return msg;
        }
        #endregion
    }
}
